package com.digiwin.demo.util;

import com.digiwin.demo.constant.ConfigConstant;
import com.digiwin.demo.mq.MyRMQSource;
import com.rabbitmq.client.BuiltinExchangeType;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;

import java.io.Serializable;

/**
 * @Author yanggld
 * @Date 2019/12/26-14:17
 * @Description
 */
public class RMQUtil<T>{

	private static final String EXCHANGE_NAME = ConfigConstant.RMQ_EXCHANGE_NAME;
	private static final String RMQ_HOST = ConfigConstant.RMQ_HOST;
	private static final int PORT = ConfigConstant.RMQ_PORT;
	private static final String RMQ_USERNAME = ConfigConstant.RMQ_USERNAME;
	private static final String RMQ_PASSWORD = ConfigConstant.RMQ_PASSWORD;

	private String routingKey;
	private String queueName;

	public RMQUtil(String routingKey,String queueName) {
		this.routingKey = routingKey;
		this.queueName = queueName;
	}

	public DataStream<T> getDataStream(StreamExecutionEnvironment env) {
		RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
				.setHost(RMQ_HOST)   //地址
				.setPort(PORT)
				.setUserName(RMQ_USERNAME)     //别用默认的，自己创建一个用户，注意用户权限
				.setPassword(RMQ_PASSWORD)
				.setVirtualHost("/")
				.build();
		//2.1. rabbitmq连接的配置，2.rabbitmq的队列名，消费的队列名
		MyRMQSource<T> myRMQSource = new MyRMQSource<>(connectionConfig,
				queueName, true, new SimpleStringSchema());
		myRMQSource.setExchangeName(EXCHANGE_NAME);
		myRMQSource.setExchangeType(BuiltinExchangeType.TOPIC);
		myRMQSource.setRoutingKey(routingKey);
		DataStream<T> dataStreamSource = env.addSource(myRMQSource);
		return dataStreamSource;
	}
}
